/*
This file is part of [aconnect] library. 

Author: Artem Kustikov (kustikoff[at]tut.by)


This code is provided 'as-is', without any express or implied
warranty. In no event will the authors be held liable for any
damages arising from the use of this code.

Permission is granted to anyone to use this code for any
purpose, including commercial applications, and to alter it and
redistribute it freely, subject to the following restrictions:

1. The origin of this code must not be misrepresented; you must
not claim that you wrote the original code. If you use this
code in a product, an acknowledgment in the product documentation
would be appreciated but is not required.

2. Altered source versions must be plainly marked as such, and
must not be misrepresented as being the original code.

3. This notice may not be removed or altered from any source
distribution.
*/

#ifndef ACONNECT_SERVER_H
#define ACONNECT_SERVER_H

#include "thread_pool.hpp"

#include "server_settings.hpp"

namespace aconnect 
{
	typedef boost::function1<void, const socket_type> process_error_proc;
	typedef void (*server_thread_proc) (class Server *);   
	typedef void (*process_stop_proc) ();
	
	//==================================================================//
	//	struct ClientInfo
	//==================================================================//
	struct ClientInfo
	{
		port_type				port;
		ip_addr_type			ip;
		mutable socket_type		sock;
		class Server			*server;

		ClientInfo() {
			reset();
		}

		ClientInfo(const ClientInfo &other)
		{
			port = other.port;
			sock = other.sock;
			memcpy (ip, other.ip, sizeof (ip_addr_type));
			server = other.server;
		}


		ClientInfo& operator= (const ClientInfo& other)	{
			ClientInfo copy(other);
			swap(copy);
			return *this; 
		}
				
		inline void reset()
		{
			port = 0; 
			sock = INVALID_SOCKET;
			server = NULL;
			util::zeroMemory(ip, sizeof(ip));
		}

		inline bool isClosed () const { return this->sock == INVALID_SOCKET; }
		
		inline void swap (ClientInfo& other) {
			
			if (&other == this)
				return;

			static const int IpAddrSize = ARRAY_SIZE(ip);

			std::swap (port, other.port);
			std::swap (sock, other.sock);
			
			for (int ndx=0; ndx<IpAddrSize; ++ndx)
				std::swap (ip[ndx], other.ip[ndx]);
			
			std::swap (server, other.server);
		};

		//==================================================================//
		//	Utility methods
		//==================================================================//

		string getRequest (SocketStateCheck &stateCheck) const;
		void writeResponse (string_constref response) const;
		
		void close (bool closeSocket = true, bool throwError = true) const throw (socket_error);
	};
	
  
	
	//==================================================================//
	//	class Server													//
	//==================================================================//
	class Server : public IStoppable, private boost::noncopyable
	{
		typedef ThreadPool<ClientInfo> pool_type;

    public:
		Server (Logger* log = 0) : 
			_threadPool (),
			_port (-1), 
			
			_errorProcess_proc (0),
			_stopProcess_proc (0),
			_mainThread (0), 
			_socket (INVALID_SOCKET),
			
			_logger (log),
			_isStopped (false)
		{ }

		virtual ~Server () { 
			clear ();
		}

		inline void init (port_type port, 
			pool_type::worker_thread_proc workerProc, 
			const ServerSettings &settings = ServerSettings())
		{
			_port = port;
			_settings = settings;

			_threadPool.setLog (_logger);
			_threadPool.setWorkerProc (workerProc);
			_threadPool.setMaxWorkersCount (settings.workersCount);

			setPoolCallbacks (Server::checkProcessedClientState, Server::checkQueuedClientState);
		}

		inline void process (ClientInfo &client) {
			_threadPool.process (client);
		}

		
		void start (bool inCurrentThread = false) throw (socket_error);
		void stop (bool waitAllWorkers = false);
		
		// server thread function
		void static run (Server *server);

		// server stop utility
		void static stopPool (Server *server);

		inline void join ()
        {
			if (_mainThread)
				_mainThread->join();
		}
		

        //==================================================================//
        //      Properties
		//==================================================================//
		virtual bool isStopped ()							{   return _isStopped;      }   

		inline port_type port()	const						{	return _port;		    }
		inline socket_type socket() const					{	return _socket;		    }
		inline const ServerSettings& settings()	const		{	return _settings;	    } 
		inline process_error_proc errorProcessProc() const	{	return _errorProcess_proc;	} 
		inline process_stop_proc stopProcessProc() const	{	return _stopProcess_proc;	} 
		
		inline void setLog (Logger *log)					{   _logger = log; _threadPool.setLog (_logger);	}
        inline Logger* log () const							{   return _logger;									}   
		
				
		inline void setErrorProcessProc(process_error_proc proc)	{	_errorProcess_proc = proc;} 
		inline void setStopProcessProc(process_stop_proc proc)		{	_stopProcess_proc = proc;} 

        inline pool_type::worker_thread_proc workerProc() const { return _threadPool.workerProc ();} 

		inline void setPoolCallbacks (pool_type::check_item_proc processedItemProc = 0, 
			pool_type::check_item_proc queuedItemProc = 0) 
		{	
			_threadPool.setProcessedItemProc (processedItemProc);
			_threadPool.setQueuedItemProc (queuedItemProc);
		}


		inline long currentWorkersCount () {	
			return _threadPool.workersCount ();
		}
		
		inline long currentPendingWorkersCount () {	
			return _threadPool.pendingWorkersCount ();
		}

		inline size_t requestsQueueLength () {	
			return _threadPool.requestsQueueLength();
		}
			
		inline void logDebug (string_constptr format, ...)	{	
			if (_logger) {
				FORMAT_VA_MESSAGE (format, formattedMessage);
				_logger->debug (formattedMessage.c_str());
			}
		}
		inline void logWarning (string_constptr format, ...)	{	
			if (_logger) {
				FORMAT_VA_MESSAGE (format, formattedMessage);
				_logger->warn (formattedMessage.c_str());
			}
		}
		inline void logWarning (const std::exception &err) {
			if (_logger)
				_logger->warn(err);
		}
		inline void logError( string_constptr format, ...)	{	
			if (_logger) {
				FORMAT_VA_MESSAGE (format, formattedMessage);
				_logger->error (formattedMessage.c_str());
			}
		}
		inline void logError (const std::exception &err) {
			if (_logger)
				_logger->error(err);
		}

		
        //==================================================================//
        //      Default pool callbacks										//
		//==================================================================//
		static void checkProcessedClientState (const aconnect::ClientInfo& client);
		static void checkQueuedClientState (const aconnect::ClientInfo& client);
		//==================================================================//

	protected:
		void applySettings ();
		
		
		inline void clear () {
			if (_mainThread) {
				delete _mainThread; 
				_mainThread = NULL;
			}
		}

	
	// fields
	protected:
		pool_type _threadPool;

		port_type _port;
        
		process_error_proc  _errorProcess_proc;
		process_stop_proc   _stopProcess_proc;

		ServerSettings _settings;
        
        boost::thread *_mainThread;
        socket_type _socket;
        
		Logger     *_logger;

       	boost::mutex _loadMutex;
		boost::condition _stopCondition;

		bool _isStopped;
	};

	//==================================================================//
}

namespace std 
{
	template<>
	inline void swap(aconnect::ClientInfo& one, aconnect::ClientInfo& other)
	{
		one.swap(other);
	}
} 


#endif // ACONNECT_SERVER_H
